package org.apache.hadoop.hive.ql.hooks;

import com.sql.blood.common.util.GraphDatabaseConnection;
import com.sql.blood.common.model.TabLine;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.ql.lib.*;
import org.apache.hadoop.hive.ql.parse.*;
import java.io.IOException;
import java.util.*;

public class HiveParse implements NodeProcessor {

    private String url;
    private TreeSet<String> inputTableList = new TreeSet<String>();
    private TreeSet<String> outputTableList = new TreeSet<String>();

    void setUrl(String url) { this.url = url; }

    /**
     * Implements the process method for the NodeProcessor interface.
     */
    public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, Object... nodeOutputs) throws SemanticException {
        ASTNode pt = (ASTNode) nd;

        switch (pt.getToken().getType()) {

            case HiveParser.TOK_CREATETABLE:
            case HiveParser.TOK_TAB:
                String table1 = BaseSemanticAnalyzer.getUnescapedName((ASTNode) pt.getChild(0));
                String outputTable = HiveUtils.fillDB(table1);
                outputTableList.add(outputTable);
                break;

            case HiveParser.TOK_TABREF:
                ASTNode tabTree = (ASTNode) pt.getChild(0);
                String table2 = (tabTree.getChildCount() == 1) ?
                        BaseSemanticAnalyzer.getUnescapedName((ASTNode) tabTree.getChild(0)) :
                        BaseSemanticAnalyzer.getUnescapedName((ASTNode) tabTree.getChild(0)) + "." + tabTree.getChild(1);
                String inputTable = HiveUtils.fillDB(table2);
                inputTableList.add(inputTable);
                break;
        }
        return null;
    }

    /**
     * parses given query and gets the lineage info.
     *
     * @param query
     * @throws ParseException
     */
    void getLineageInfo(String query) throws ParseException, SemanticException {

        System.out.println("parse sql: " + query);

        /*
         * Get the AST tree
         */
        ParseDriver pd = new ParseDriver();
        ASTNode tree = pd.parse(query);

        while ((tree.getToken() == null) && (tree.getChildCount() > 0)) {
            tree = (ASTNode) tree.getChild(0);
        }

        /*
         * initialize Event Processor and dispatcher.
         */
        inputTableList.clear();
        outputTableList.clear();

        // create a walker which walks the tree in a DFS manner while maintaining
        // the operator stack. The dispatcher
        // generates the plan from the operator tree
        Map<Rule, NodeProcessor> rules = new LinkedHashMap<Rule, NodeProcessor>();

        // The dispatcher fires the processor corresponding to the closest matching
        // rule and passes the context along
        Dispatcher disp = new DefaultRuleDispatcher(this, rules, null);
        GraphWalker ogw = new DefaultGraphWalker(disp);

        // Create a list of topop nodes
        ArrayList<Node> topNodes = new ArrayList<Node>();
        topNodes.add(tree);
        ogw.startWalking(topNodes, null);

        System.out.println("input table : " + StringUtils.join(inputTableList,","));
        System.out.println("output table : " + StringUtils.join(outputTableList,","));
    }

    List<TabLine> getInputTableList() {
        List<TabLine> tabLines = new ArrayList<TabLine>();
        for (String bd_table:inputTableList){
            TabLine tabLine = HiveUtils.splitDbTable(url, bd_table);
            tabLines.add(tabLine);
        }
        return tabLines;
    }

    List<TabLine> getOutputTableList() {
        List<TabLine> tabLines = new ArrayList<TabLine>();
        for (String bd_table:outputTableList){
            TabLine tabLine = HiveUtils.splitDbTable(url, bd_table);
            tabLines.add(tabLine);
        }
        return tabLines;
    }

    /*public static void main(String[] args) throws IOException, ParseException, SemanticException {
        //insert into table default.liuxinpei3 select * from default.liuxinpei2
        String query = "insert into table aa select * from bb union all select * from cc";
        String query1 = "create table if not exists aaa(id string, name string)";
        String query2 = "CREATE table valid_records AS SELECT * FROM records2 WHERE temperature !=9999";
        String query3 = "delete table valid_records";
        HiveParse hiveParse = new HiveParse();
        hiveParse.getLineageInfo("create table if not exists ccc(id string, user_name string)");
        List<TabLine> input = hiveParse.getInputTableList();
        List<TabLine> output = hiveParse.getOutputTableList();
        GraphDatabaseConnection conn = new GraphDatabaseConnection();
        for (TabLine tabLine:input){
            conn.insertVertex(tabLine);
        }
        for (TabLine tabLine:output){
            conn.insertVertex(tabLine);
        }
        for (TabLine x:input){
            for (TabLine y:output){
                conn.insertEdge(x,y);
            }
        }
        conn.close();
    }*/
}
